package com.changdu.seatunnel.admin.service;

import java.time.Duration;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;

import org.apache.commons.lang3.StringUtils;
import org.springframework.stereotype.Service;
import org.springframework.web.client.RestTemplate;

import com.changdu.seatunnel.admin.config.common.BusinessException;
import com.changdu.seatunnel.admin.config.enums.SyncErrorStatus;
import com.changdu.seatunnel.admin.config.enums.SyncTaskStatus;
import com.changdu.seatunnel.admin.pojo.SeatunnelTaskInfo;
import com.changdu.seatunnel.admin.util.JsonUtils;

import lombok.extern.slf4j.Slf4j;

/**
 * 
 * @author wangxiaoliang
 * @param <V>
 */
@Slf4j
@Service
@SuppressWarnings("unchecked")
public class SeatunnelRester {

	private static final String SUBMIT = "hazelcast/rest/maps/submit-job";
	private static final String SUBMIT_SUB = "hazelcast/rest/maps/submit-subjob";
	private static final String STOP = "hazelcast/rest/maps/stop-job";
	private static final String INFO = "hazelcast/rest/maps/job-info"; 
	
    private RestTemplate restTemplate = new RestTemplate();	
	
	/**
	 * 发送主任务
	 * @param centerDomain
	 * @param templateContent
	 */
	public String sendJobRequest(String centerDomain, String taskId, String templateContent) {
        try {
        	String url = String.format("%s/%s", centerDomain, SUBMIT);
        	if (Objects.nonNull(taskId)) {
        		url = String.format("%s?jobId=%s&isStartWithSavePoint=true", url, taskId);
        	}
        	log.info("send url: {}, submit task: {}", url, templateContent);
        	Map<String, Object> response = restTemplate.postForObject(url, templateContent, Map.class);
        	log.info("accept url: {}, submit task: response: {}", url, response);
        	return String.valueOf(response.get("jobId").toString());
        } catch (Exception e) {
        	log.error(e.getMessage(), e);
        	throw new BusinessException(SyncErrorStatus.SUBMIT_EXCEPTION.getCode(), e.getMessage());
        }
	}
	
	/**
	 * 发送子任务
	 * 
	 * @param centerDomain
	 * @param templateContent
	 */
	public void sendKafksSubRequest(String centerDomain, String templateContent) {
        try {
        	String url = String.format("%s/%s", centerDomain, SUBMIT_SUB);
        	log.info("send url: {}, kafks subtask: {}", url, templateContent);
        	Map<String, Object> response = restTemplate.postForObject(url, templateContent, Map.class);
        	log.info("accept url: {}, kafks subtask response: {}", url, response);
        } catch (Exception e) {
        	log.error(e.getMessage(), e);
        	throw new BusinessException(SyncErrorStatus.SUBMIT_EXCEPTION.getCode(), e.getMessage());
        }
	}
	
	/**
	 * 停止任务
	 * 
	 * @param centerDomain
	 * @param templateContent
	 */
	public void sendStopRequest(String centerDomain, String taskId, boolean savepoint) {
		if (StringUtils.isEmpty(taskId)) return;
		
		HashMap<String, Object> args = new HashMap<>();
		args.put("jobId", taskId);
		args.put("isStopWithSavePoint", savepoint);
		
        try {
        	String url = String.format("%s/%s", centerDomain, STOP);
        	log.info("send url: {}, stop task: {}", url, taskId);
        	Map<String, Object> response = restTemplate.postForObject(url, JsonUtils.toJson(args), Map.class);
        	log.info("accept url: {}, stop task response: {}", url, response);
        } catch (Exception e) {
        	log.error(e.getMessage(), e);
        	throw new BusinessException(SyncErrorStatus.SUBMIT_EXCEPTION.getCode(), e.getMessage());
        }
	}
	
	/**
	 * 恢复任务
	 * 
	 * @param centerDomain
	 * @param templateContent
	 */
	public void sendDeleteRequest(String centerDomain, String taskId) {
		HashMap<String, Object> args = new HashMap<>();
		args.put("jobId", taskId);
		args.put("isStopWithSavePoint", false);
		
        try {
        	String url = String.format("%s/%s", centerDomain, STOP);
        	log.info("send url: {}, delete task: {}", url, taskId);
        	Map<String, Object> response = restTemplate.postForObject(url, JsonUtils.toJson(args), Map.class);
        	log.info("accept url: {}, delete task response: {}", url, response);
        } catch (Exception e) {
        	log.error(e.getMessage(), e);
        	throw new BusinessException(SyncErrorStatus.SUBMIT_EXCEPTION.getCode(), e.getMessage());
        }
	}

	/**
	 * 查询任务状态信息
	 * @param domain
	 * @param taskId
	 * @return
	 */
	public SeatunnelTaskInfo getSyncTaskStatus(String centerDomain, String taskId) {
		try {
        	String url = String.format("%s/%s/%s", centerDomain, INFO, taskId);
        	log.info("send url: {}, task info: {}", url, taskId);
        	Map<String, Object> response = restTemplate.getForObject(url, Map.class);
        	log.info("accept url: {}, task info response: {}", url, response);
        	
        	SeatunnelTaskInfo taskInfo = new SeatunnelTaskInfo();
        	taskInfo.setJobId(String.valueOf(response.get("jobId")));
        	if (Objects.nonNull(response.get("jobName"))) {
        		taskInfo.setJobName(String.valueOf(response.get("jobName")));
        		taskInfo.setJobStatus(String.valueOf(response.get("jobStatus")));
        		taskInfo.setCreateTime(String.valueOf(response.get("createTime")));
        		
        		Map<String, Object> metrics = (Map<String, Object>) response.get("metrics");
        		taskInfo.setSourceReceivedCount(Long.valueOf(metrics.get("SourceReceivedCount").toString()));
        		taskInfo.setSourceReceivedQPS(Double.valueOf(metrics.get("SourceReceivedQPS").toString()));
        		taskInfo.setSinkWriteCount(Long.valueOf(metrics.get("SinkWriteCount").toString()));
        		taskInfo.setSinkWriteQPS(Double.valueOf(metrics.get("SinkWriteQPS").toString()));
        	} else {
        		taskInfo.setJobStatus(SyncTaskStatus.NOT_EXISTS.getSeatunnelStatus());
        	}
        	return taskInfo;
        } catch (Exception e) {
        	log.error(e.getMessage(), e);
        	throw new BusinessException(SyncErrorStatus.SUBMIT_EXCEPTION.getCode(), e.getMessage());
        }		
	}
	
}
